package com.fitz.cluster;

import com.alibaba.fastjson.JSON;
import com.fitz.common.LocalCache;
import com.fitz.invoke.RpcInvoker;
import com.fitz.proxy.RpcReferenceBean;
import com.fitz.registry.InstanceInfo;
import com.fitz.remote.callback.RpcResponseCallback;
import com.fitz.remote.client.ConsumerClient;
import com.fitz.remote.client.pool.ConsumerClientPool;
import com.fitz.remote.client.pool.InitClientPool;
import com.fitz.remote.model.RpcRequest;
import com.fitz.remote.model.RpcResponse;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author fitz
 */

public abstract class AbstractClusterStrategy {

    private static final Logger logger = LoggerFactory.getLogger(AbstractClusterStrategy.class);

    public abstract RpcResponse call(RpcRequest rpcRequest, RpcReferenceBean referenceBean);

    /**
     * 根据负载均衡策略发现服务地址
     * @param referenceBean
     * @return
     */
    private InstanceInfo discovery(RpcReferenceBean referenceBean){
        if (referenceBean.getInstance()!=null){
            logger.debug("discovery default address :{}", JSON.toJSONString(referenceBean.getInstance()));
            return referenceBean.getInstance();
        }
        InstanceInfo instance = referenceBean.getInstanceRegistry().discover(referenceBean.getInstanceRegistry(),
                referenceBean.getLoadBalance(),referenceBean.getInterfaceName());
        logger.debug("discovery :{}",JSON.toJSONString(instance));
        return instance;
    }

    ConsumerClient getClient(RpcReferenceBean referenceBean){
        InstanceInfo instance = discovery(referenceBean);
        String serverKey = instance.serverKey();
        ConsumerClientPool clientPool = LocalCache.getClientPoolMap().get(serverKey);
        if (clientPool == null){
            logger.info("clientPool is null,now begin to init");
            List<InstanceInfo> serverList = referenceBean.getInstanceRegistry().getAllInstances(referenceBean.getInterfaceName());
            InitClientPool.initClientPool(referenceBean.getConnectPoolParam().getMinPoolSize(),
                    referenceBean.getConnectPoolParam().getMaxPoolSize(),
                    referenceBean.getConnectPoolParam().getSerializerType(),
                    serverList);
        }
        clientPool = LocalCache.getClientPoolMap().get(serverKey);
        ConsumerClient client = clientPool.getClient();
        client.setInstance(instance);
        return client;
    }

    void returnClient(ConsumerClient client){
        if (client == null){
            return;
        }
        InstanceInfo instance = client.getInstance();
        ConsumerClientPool clientPool = LocalCache.getClientPoolMap().get(instance.serverKey());
        clientPool.returnClient(client);
    }

    /**
     * @desc: 发送rpc请求
     * @author: fitz
     * @Date: 2020/5/25
     * @param: [rpcRequest, serverAddress]
     * @return: com.fitz.remote.model.RpcResponse
     **/
    RpcResponse sendRequest(RpcRequest rpcRequest,ConsumerClient client){
        RpcResponseCallback rpcResponseCallback = new RpcResponseCallback(RpcInvoker.getInstance(),rpcRequest,client);
        try {
            return rpcResponseCallback.sendRequest();
        } finally {
            rpcResponseCallback.remove();
        }
    }
}
